package zisu.netty.practice.RPC.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;
import zisu.netty.practice.RPC.client.codec.RpcDecode;
import zisu.netty.practice.RPC.client.codec.RpcEncode;
import zisu.netty.practice.RPC.client.codec.RpcRequest;
import zisu.netty.practice.RPC.client.codec.RpcResponse;
import zisu.netty.practice.RPC.config.provider.ProviderConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @Desc: ""
 * @Author: caixiang
 * @DATE: 2021/1/20 9:25
 */
@Slf4j
public class RpcServer {
    private String serverAddress;

    //server 有两个 EventLoopGroup，
    //bossLoopGroup是用于接收连接的。
    private EventLoopGroup bossLoopGroup = new NioEventLoopGroup();
    //workGroup 是用来处理事件的group
    private EventLoopGroup workLoopGroup = new NioEventLoopGroup();

    private volatile Map<String /* interface name */,Object /* 调用的接口 */> handlerMap = new HashMap<>();


    public RpcServer(String serverAddress) throws InterruptedException {
        this.serverAddress = serverAddress;
        this.start();
    }

    private void start() throws InterruptedException {
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossLoopGroup,workLoopGroup)
                .channel(NioServerSocketChannel.class)
                //TCP = sync(队列) + accept(队列) = so_backlog（1024）
                .option(ChannelOption.SO_BACKLOG,1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,4,0,0));
                        //服务端接受的是 request对象
                        pipeline.addLast(new RpcDecode(RpcRequest.class));
                        //服务端反馈的是 response对象
                        pipeline.addLast(new RpcEncode(RpcResponse.class));
                        pipeline.addLast(new RpcServerHandler(handlerMap));
                    }
                });
        String[] split = serverAddress.split(":");
        String host = split[0];
        Integer port = Integer.parseInt(split[1]);
        ChannelFuture channelFuture = serverBootstrap.bind(host, port).sync();
        channelFuture.addListener(new ChannelFutureListener() {
            //channelFuture 内的listener 是异步的
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if(future.isSuccess()){
                    log.info("server success bing to :"+ serverAddress);
                }else {
                    log.info("server fail bing to :"+ serverAddress);
                    throw new Exception("server start fail :"+future.cause());
                }
            }
        });

        //这里的代码 是同步的。
        try {
            //等待 知道5秒后打断 这个channelFuture就行了。
            channelFuture.awaitUninterruptibly(5000, TimeUnit.MILLISECONDS);
            if(channelFuture.isSuccess()){
                log.info("start rapid rpc success!");
            }
        }catch (Exception e){
            log.error("start rapid rpc occur Interrupt, e:"+e);
        }
    }

    /**
     * registerProcessor 程序注册器
     * */
    public void registerProcessor(ProviderConfig providerConfig){
        //key :  providerConfig.interface() (userService接口权限命名) ;;
        //value :  providerConfig.ref (userService 下的具体实现类 userServiceImpl实例对象) ;;

        handlerMap.put(providerConfig.getInterface(),providerConfig.getRef());
    }

    /**
     * registerProcessor 程序注册器
     * */
    public void close(){
        bossLoopGroup.shutdownGracefully();
        workLoopGroup.shutdownGracefully();
    }

}
